//******************************************************************************************************
// Copyright 2008 Amazon Technologies, Inc.  
// Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in 
// compliance with the License. 
//
// You may obtain a copy of the License at:http://aws.amazon.com/apache2.0  This file is distributed on 
// an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 
//
// See the License for the specific language governing permissions and limitations under the License. 
//******************************************************************************************************using System;
using System;
using System.Collections;
using System.ComponentModel;
using System.Data;
using System.Text;
using System.Windows.Forms;
using System.Threading;

//
// References for SQS operations...
//
using Amazon.SQS;
using Amazon.SQS.Mock;
using Amazon.SQS.Model;
using SQS_Client;

namespace SQS
{
    public abstract class SQSManager : Component
    {
        public enum pollMode
        {
            Adaptive,
            Manual
        }
        
        //
        // Another thread could potentially change these values so mark them as volatile
        //
        static volatile int    _threadsAllocated = 0;
    
        //
        // Component statistics...
        //
        volatile int    _receiveTimeouts       = 0;
        volatile int    _receiveErrors         = 0;
        volatile int    _processingErrors      = 0;

        volatile int    _requestMS             = 0;
        volatile int    _requestsProcessed     = 0;

        volatile int    _sqsMS                 = 0;

        long            _receives              = 0;
        long            _deletes               = 0;
        long            _getMsgCount           = 0;
        long            _messageCount          = -1;

        bool            _measureNetLatency     = false;
        int             _networkLatency        = -1;
        RouteTrace      _routeTrace            = new RouteTrace();
    
        string          _awsID                 = null;
        string          _awsSecretKey          = null;

        string          _queue                 = "sqsWorker";
        bool            _autoDelete            = false;    // auto delete bad messages
        
        int             _threads               = 1;
        int             _receiveDelay          = 500;

        pollMode        _pollingMode           = pollMode.Adaptive;
    
        bool            _running               = false;
    
        string          _pollQueue             = null;
        ArrayList       _dequeueThreads        = ArrayList.Synchronized(new ArrayList());
        Queue           _threadCleanup         = System.Collections.Queue.Synchronized(new System.Collections.Queue());
        Thread          _messageCountThread    = null;
        Thread          _netLatencyThread      = null;
        EventWaitHandle _stopThreads           = new EventWaitHandle(false, EventResetMode.ManualReset);
        EventWaitHandle _stopMessageCount      = new EventWaitHandle(false, EventResetMode.ManualReset);

        #region Designer Properties
		[Bindable( true ), Category( "Display" ), NotifyParentProperty( true ),
		DefaultValue( 1 ),
		Description( "Number of Threads" )]
		public int Threads
		{
			get { return _threads; }
			set { _threads = value; }
		}
    
		[Bindable( true ), Category( "Display" ), NotifyParentProperty( true ),
		DefaultValue( 1 ),
		Description( "Automatically delete message that cause processing failures" )]
		public bool AutoDelete
		{
			get { return _autoDelete; }
			set { _autoDelete = value; }
		}

        [Bindable( true ), Category( "Display" ), NotifyParentProperty( true ),
		DefaultValue( 500 ),
		Description( "Receive poll delay (ms)" )]
		public int ReceiveDelay
		{
			get { return _receiveDelay; }
			set { _receiveDelay = value; }
		}
    
		/// <summary>
		/// </summary>
		[Bindable( true ), Category( "Display" ), NotifyParentProperty( true ),
		DefaultValue( "Queue Name" ),
		Description( "SQS Worker Request Queue" )]
		public string Queue
		{
			get { return _queue; }
			set { _queue = value; }
		}
    
		/// <summary>
		/// </summary>
		[Bindable( true ), Category( "Display" ), NotifyParentProperty( true ),
		DefaultValue( null ),
		Description( "AWS ID" )]
		public string AwsID
		{
			get { return _awsID; }
			set { _awsID = value; }
		}
    
		/// <summary>
		/// </summary>
		[Bindable( true ), Category( "Display" ), NotifyParentProperty( true ),
		DefaultValue( null ),
		Description( "AWS SecretKey" )]
		public string AWSSecretKey
		{
			get { return _awsSecretKey; }
			set { _awsSecretKey = value; }
		}
    
		/// <summary>
		/// </summary>
		[Bindable( true ), Category( "Display" ), NotifyParentProperty( true ),
		DefaultValue( null ),
		Description( "Measure Network Latency" )]
		public bool MeasureNetLatency
		{
			get { return _measureNetLatency; }
			set { _measureNetLatency = value; }
		}
    
		/// <summary>
		/// </summary>
		[Bindable( true ), Category( "Display" ), NotifyParentProperty( true ),
		DefaultValue( null ),
		Description( "Type of polling alogrithm" )]
		public pollMode PollingMode
		{
			get { return _pollingMode; }
			set { _pollingMode = value; }
		}
        #endregion Designer Properties
        #region Properties
		/// <summary>
		/// </summary>
        [Bindable( false ), Browsable( false ),
        DesignerSerializationVisibility( DesignerSerializationVisibility.Hidden )]
        public bool Running
		{
			get { return _running; }
		}
        [Bindable( false ), Browsable( false ),
        DesignerSerializationVisibility( DesignerSerializationVisibility.Hidden )]
		public int ReceiveTimeOuts
		{
			get { return _receiveTimeouts; }
		}
		/// <summary>
		/// </summary>
        [Bindable( false ), Browsable( false ),
        DesignerSerializationVisibility( DesignerSerializationVisibility.Hidden )]
        public int ReceiveErrors
		{
			get { return _receiveErrors; }
		}
		/// <summary>
		/// </summary>
        [Bindable( false ), Browsable( false ),
        DesignerSerializationVisibility( DesignerSerializationVisibility.Hidden )]
        public int ProcessingErrors
		{
			get { return _processingErrors; }
		}

        /// <summary>
		/// </summary>
        [Bindable( false ), Browsable( false ),
        DesignerSerializationVisibility( DesignerSerializationVisibility.Hidden )]
        public int RequestsProcessed
		{
			get { return _requestsProcessed; }
		}
        /// <summary>
		/// </summary>
        [Bindable( false ), Browsable( false ),
        DesignerSerializationVisibility( DesignerSerializationVisibility.Hidden )]
        public double AverageProcessMS
		{
			get 
            { 
                double   result = 0.0;
                if (_requestsProcessed > 0)
                    result = (double)_requestMS / (double)_requestsProcessed;
                return result; 
            }
		}
        /// <summary>
		/// </summary>
        [Bindable( false ), Browsable( false ),
        DesignerSerializationVisibility( DesignerSerializationVisibility.Hidden )]
        public long MessageCount
		{
			get { return _messageCount; }
		}
        /// <summary>
		/// </summary>
        [Bindable( false ), Browsable( false ),
        DesignerSerializationVisibility( DesignerSerializationVisibility.Hidden )]
        public long Receives
		{
			get { return _receives; }
		}
        /// <summary>
		/// </summary>
        [Bindable( false ), Browsable( false ),
        DesignerSerializationVisibility( DesignerSerializationVisibility.Hidden )]
        public long Deletes
		{
			get { return _deletes; }
		}
        /// <summary>
		/// </summary>
        [Bindable( false ), Browsable( false ),
        DesignerSerializationVisibility( DesignerSerializationVisibility.Hidden )]
        public long GetMessageCount
		{
			get { return _getMsgCount; }
		}
        [Bindable( false ), Browsable( false ),
        DesignerSerializationVisibility( DesignerSerializationVisibility.Hidden )]
        public double AverageLatency
		{
			get 
            { 
                double latency = -1.0;
                if (_receives + _deletes + _getMsgCount > 0.0)
                    latency = (double)_sqsMS / (double)(_receives + _deletes + _getMsgCount);
                return latency; 
            }
		}
        [Bindable( false ), Browsable( false ),
        DesignerSerializationVisibility( DesignerSerializationVisibility.Hidden )]
        public int NetworkLatency
		{
			get { return _networkLatency; }
		}
        [Bindable( false ), Browsable( false ),
        DesignerSerializationVisibility( DesignerSerializationVisibility.Hidden )]
        public int ActiveThreads
		{
			get { return _dequeueThreads.Count; }
		}

        #endregion Properties
        #region Public Methods
    
        public bool Start()
        {
            if (_running) return false;
        
            _stopThreads.Reset();
            _stopMessageCount.Reset();
            _threadCleanup.Clear();
            _receiveTimeouts    = 0;
            _receiveErrors      = 0;
            _processingErrors   = 0;
            _requestMS          = 0;
            _sqsMS              = 0;
            _requestsProcessed  = 0;

            _messageCount       = -1;
            _receives            = 0;
            _deletes             = 0;
            _getMsgCount       = 0;

            _networkLatency     = -1;

            _threadsAllocated  += _threads + 2; // Threads plus message count thread plus
                                                 //  potentially traceroute thread...
        
            Client.Startup(AwsID.Trim(), AWSSecretKey.Trim());
            Client.SetConnectionLimit(_threadsAllocated);
            
            //
            // Ensure that queues exist...
            //
            _pollQueue = ResolvePollQueue();
            Client.CreateQueue(_pollQueue);
        
            switch (PollingMode)
            {
                case pollMode.Adaptive:
                    AdaptiveStart();
                    break;
                case pollMode.Manual:
                    ManualStart();
                    break;
            }
        
            //
            // Determine if we should start network latency measurement thread...
            //
            if (_measureNetLatency)
            {
                _netLatencyThread   = new Thread(new ThreadStart(TraceRoute));
                _netLatencyThread.Start();
            }
            
            _running    = true;
         
            return true;
        }
    
        public bool Stop()
        {
            if (_running == false) return false;
        
            _stopThreads.Set();
            _stopMessageCount.Set();
        
            for (int i = 0; i < _dequeueThreads.Count; i++)
            {
                Thread t = (Thread)_dequeueThreads[i];
                if (t != null && t.Join(2000) == false) t.Abort();
            }
            _dequeueThreads.Clear();
        
            if (_messageCountThread != null)
            {
                if (_messageCountThread.Join(2000) == false)
                    _messageCountThread.Abort();
                _messageCountThread = null;
            }

            if (_netLatencyThread != null)
            {
                if (_netLatencyThread.Join(2000) == false)
                    _netLatencyThread.Abort();
                _netLatencyThread = null;
            }
        
            _threadsAllocated  -= _threads;
            _running             = false;
            return true;
        }
    
        #endregion Public Methods
        #region Private Methods
    
        private void ManualStart()
        {
            for (int i = 0; i < _threads; i++)
            {
                Thread t = new Thread(new ThreadStart(ManualDequeue));
                _dequeueThreads.Add(t);

                t.Start();
            }
        
            _messageCountThread = new Thread(new ThreadStart(RetrieveMsgCount));
            _messageCountThread.Start();
        }
    
        private void AdaptiveStart()
        {        
            Thread t = new Thread(new ThreadStart(AdaptiveDequeue));
            _dequeueThreads.Add(t);
            t.Start();
        }
    
        private void AdaptiveDequeue()
        {
            Thread  t;
            int     delay;
            int     emptyReceiveCount = 0;
        
            while (!_stopThreads.WaitOne(0, false))
            {
                //           
                // Determine if we processed a message...
                //
                if (PolllSQS(true)) 
                {   
                    //
                    // Processed message
                    //

                    //
                    // Make sure message count thread is running...                
                    //
                    emptyReceiveCount = 0;                
                    delay = 1;                  // processed a message, try again quickly
                }
                else                    
                {   //
                    // No message
                    //

                    delay = _receiveDelay;      // did not process a message, so delay
                    emptyReceiveCount++;

                    if (emptyReceiveCount > 20 && _messageCount == 0)
                    {
                        _stopMessageCount.Set();
                        _messageCountThread = null;
                    }
                }
            
                //
                // First - cleanup dead threads...
                //
                while (_threadCleanup.Count > 0)
                {
                    t = (Thread)_threadCleanup.Dequeue();
                    _dequeueThreads.Remove(t);
                }
            
                Thread.Sleep(delay);
            }
        }
    
        private void AdaptiveThreadCheck()
        {
            if (_messageCountThread == null)
            {
                _stopMessageCount.Reset();
                _messageCount     = -1;
                _messageCountThread = new Thread(new ThreadStart(RetrieveMsgCount));
                _messageCountThread.Start();
            }
            
            //
            // Do we need to spawn more threads...
            //
            if (_dequeueThreads.Count < _threads)
            {
                Thread t = new Thread(new ParameterizedThreadStart(AdaptiveDequeueChild));
                _dequeueThreads.Add(t);
                t.Start(t);
            }
        }
    
        private void AdaptiveDequeueChild(object t)
        {
            int delay;
            int emptyReceiveCount = 0;
            while (!_stopThreads.WaitOne(0, false))
            {
                //
                // Determine if we processed a message...
                //
                if (PolllSQS(true)) 
                {   // Processed message
                
                    //
                    // Reset empty receive count...
                    //
                    emptyReceiveCount = 0;
                    delay             = 1;        // processed a message, try again quickly
                }
                else 
                {   // No message
                    delay = _receiveDelay;
                    emptyReceiveCount++;

                    if (emptyReceiveCount > 10) break;
                }            
                Thread.Sleep(delay);
            }

            if (emptyReceiveCount > 10) _threadCleanup.Enqueue(t);
        }
    
        private void ManualDequeue()
        {
            int delay;
            while (!_stopThreads.WaitOne(0, false))
            {
                if (PolllSQS(false)) delay = 1;
                               else delay = _receiveDelay;
                Thread.Sleep(delay);
            }
        }
    
        private bool PolllSQS(bool adaptive)
        {
            Amazon.SQS.Model.Message    msg = null;
            DateTime                    timestamp;
            TimeSpan                    ts;
            bool                        rc = false;
        
            try
            {
                timestamp   = DateTime.Now;
                msg         = Client.ReceiveMessage(_pollQueue);
                ts          = DateTime.Now.Subtract(timestamp);
                _sqsMS    += Convert.ToInt32(ts.TotalMilliseconds);
                _receives++;
            }
            catch (Exception ex)
            {
                if (ex is TimeoutException) _receiveTimeouts++;
                                       else _receiveErrors++;
                return false;
            }
        
            if (msg != null)
            {
                //
                // If running adaptively, check our thread count...
                //
                if (adaptive) AdaptiveThreadCheck();
            
                //
                // Process message...
                //
                try
                {
                    timestamp    = DateTime.Now;
                    bool result  = MessageReceived(msg);
                    ts           = DateTime.Now.Subtract(timestamp);
                    
                    _requestMS += Convert.ToInt32(ts.TotalMilliseconds);
                    _requestsProcessed++;
                    
                    //
                    // Event returning true means delete message from the queue...
                    //
                    if (result == true)
                        DeleteMessage(_pollQueue, msg.ReceiptHandle);
                
                    rc = true;
                }
                catch
                {
                    _processingErrors++;
                    
                    //
                    // User has told us to automatically delete this message...
                    //
                    if (_autoDelete)
                        DeleteMessage(_pollQueue, msg.ReceiptHandle);
                }
            }
        
            return rc;
        }

        private void DeleteMessage(string q, string receipt)
        {
            DateTime timestamp = DateTime.Now;
            Client.DeleteMessage(q, receipt);
            TimeSpan ts        = DateTime.Now.Subtract(timestamp);
        
            _sqsMS  += Convert.ToInt32(ts.TotalMilliseconds);
            _deletes++;
        }
    
        private void RetrieveMsgCount()
        {            
            while (!_stopMessageCount.WaitOne(0, false))
            {
                try
                {                
                    DateTime timestamp = DateTime.Now;
                    _messageCount  = Client.ApproximateNumberOfMessages(_pollQueue);
                    TimeSpan ts = DateTime.Now.Subtract(timestamp);
        
                    _sqsMS        += Convert.ToInt32(ts.TotalMilliseconds);
                    _getMsgCount++;
                }
                catch {}
                Thread.Sleep(_receiveDelay);
            }
        }
    
        private void TraceRoute()
        {
            Uri uri = Client.Uri;
        
            do
            {
                try
                {
                    _networkLatency  = _routeTrace.Latency(uri.DnsSafeHost);
                    _networkLatency *= 2;          // account for round trip...
                }
                catch
                {
                    _networkLatency = -1;
                    break;                  // error - don't continue...
                }
            } while (!_stopThreads.WaitOne(10000, false));   // Pause 10 seconds...
        }
    
        #endregion Private Methods
        #region Overrides

        protected virtual bool MessageReceived(Amazon.SQS.Model.Message msg)
        {
            return true;
        }
    
        protected virtual string ResolvePollQueue()
        {
            return _queue;
        }
        #endregion Overrides        
    }
}
